{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ydsPE-aN1Alm"
      },
      "source": [
        "##### Copyright &copy; 2020 The TensorFlow Authors."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "uCSByGH6C7zS"
      },
      "source": [
        "<font size=-1>Licensed under the Apache License, Version 2.0 (the \\\"License\\\");\n",
        "you may not use this file except in compliance with the License.\n",
        "You may obtain a copy of the License at [https://www.apache.org/licenses/LICENSE-2.0](https://www.apache.org/licenses/LICENSE-2.0)\n",
        "\n",
        "Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an \\\"AS IS\\\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the License for the specific language governing permissions and limitations under the License.</font>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "6TyrY7lV0oke"
      },
      "source": [
        "# Create a TFX pipeline for your data with Penguin template\n",
        "\n",
        "---\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "iLYriYe10okf"
      },
      "source": [
        "## Introduction\n",
        "\n",
        "This document will provide instructions to create a TensorFlow Extended (TFX)\n",
        "pipeline for your own dataset using *penguin template* which is provided with\n",
        "TFX Python package. Created pipeline will be using\n",
        "[Palmer Penguins](https://allisonhorst.github.io/palmerpenguins/articles/intro.html)\n",
        "dataset initially,\n",
        "but we will transform the pipeline for your dataset.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "M0HDv9FAbUy9"
      },
      "source": [
        "### Prerequisites\n",
        "- Linux / MacOS\n",
        "- Python 3.6-3.8\n",
        "- Jupyter notebook\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XaXSIXh9czAX"
      },
      "source": [
        "## Step 1. Copy the predefined template to your project directory.\n",
        "In this step, we will create a working pipeline project directory and files\n",
        "by copying files from *penguin template* in TFX. You can think of this as a\n",
        "scaffold for your TFX pipeline project."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "VmVR97AgacoP"
      },
      "source": [
        "### Install required package\n",
        "First, install `tfx` python package. We fix the version of\n",
        "`tfx` and `tensorflow-model-analysis` to make sure we can have\n",
        "necessary npm dependencies for visualization.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "XNiqq_kN0okj"
      },
      "outputs": [],
      "source": [
        "# Install tfx Python package.\n",
        "import sys\n",
        "!{sys.executable} -m pip install --upgrade pip\n",
        "\n",
        "!{sys.executable} -m pip install --user --upgrade --use-feature=2020-resolver \\\n",
        "  tfx==0.25.0 tensorflow-model-analysis==0.25.0\n",
        "\n",
        "# automatically restart kernel after installs\n",
        "import IPython\n",
        "app = IPython.Application.instance()\n",
        "app.kernel.do_shutdown(True)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hX1rqpbQ0okp"
      },
      "source": [
        "Let's check the versions of TFX."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "XAIoKMNG0okq"
      },
      "outputs": [],
      "source": [
        "import tensorflow as tf\n",
        "import tensorflow_model_analysis as tfma\n",
        "import tfx\n",
        "\n",
        "print('TF version: {}'.format(tf.__version__))\n",
        "print('TFMA version: {}'.format(tfma.__version__))\n",
        "print('TFX version: {}'.format(tfx.__version__))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "TOsQbkky0ok7"
      },
      "source": [
        "We are ready to create a pipeline.\n",
        "\n",
        "Set `PROJECT_DIR` to appropriate destination for your environment.\n",
        "Default value is `~/imported/${PIPELINE_NAME}` which is appropriate for\n",
        "[Google Cloud AI Platform Notebook](https://console.cloud.google.com/ai-platform/notebooks/)\n",
        "environment.\n",
        "\n",
        "You may give your pipeline a different name by changing the `PIPELINE_NAME`\n",
        "below. This will also become the name of the project directory where your\n",
        "files will be put.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "cIPlt-700ok-"
      },
      "outputs": [],
      "source": [
        "PIPELINE_NAME=\"my_pipeline\"\n",
        "import os\n",
        "# Set this project directory to your new tfx pipeline project.\n",
        "PROJECT_DIR=os.path.join(os.path.expanduser(\"~\"), \"imported\", PIPELINE_NAME)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ozHIomcd0olB"
      },
      "source": [
        "### Copy template files.\n",
        "\n",
        "TFX includes the `penguin` template with the TFX python package.\n",
        "`penguin` template\n",
        "contains many instructions to bring your dataset into the pipeline which is\n",
        "the purpose of this tutorial.\n",
        "\n",
        "The `tfx template copy` CLI command copies predefined template files into your\n",
        "project directory."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "VLXpTTjU0olD"
      },
      "outputs": [],
      "source": [
        "# Set `PATH` to include user python binary directory and a directory containing `skaffold`.\\\n",
        "PATH=%env PATH\n",
        "%env PATH={PATH}:/home/jupyter/.local/bin\n",
        "\n",
        "!tfx template copy \\\n",
        "  --pipeline-name={PIPELINE_NAME} \\\n",
        "  --destination-path={PROJECT_DIR} \\\n",
        "  --model=penguin"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "yxOT19QS0olH"
      },
      "source": [
        "Change the working directory context in this notebook to the project directory."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "6P-HljcU0olI"
      },
      "outputs": [],
      "source": [
        "%cd {PROJECT_DIR}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "1tEYUQxH0olO"
      },
      "source": [
        ">NOTE: If you are using JupyterLab or Google Cloud AI Platform Notebook,\n",
        "don't forget to change directory in `File Browser` on the left by clicking\n",
        "into the project directory once it is created."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "IzT2PFrN0olQ"
      },
      "source": [
        "### Browse your copied source files\n",
        "\n",
        "The TFX template provides basic scaffold files to build a pipeline,\n",
        "including Python source code and sample data. The `penguin` template uses the\n",
        "same *Palmer Penguins* dataset and ML model as the\n",
        "[Penguin example](https://github.com/tensorflow/tfx/tree/master/tfx/examples/penguin).\n",
        "\n",
        "Here is brief introduction to each of the Python files.\n",
        "-   `pipeline` - This directory contains the definition of the pipeline\n",
        "    -   `configs.py` — defines common constants for pipeline runners\n",
        "    -   `pipeline.py` — defines TFX components and a pipeline\n",
        "-   `models` - This directory contains ML model definitions\n",
        "    -   `features.py`, `features_test.py` — defines features for the model\n",
        "    -   `preprocessing.py`, `preprocessing_test.py` — defines preprocessing\n",
        "         routines for data\n",
        "    -   `constants.py` — defines constants of the model\n",
        "    -   `model.py`, `model_test.py` — defines ML model using ML frameworks\n",
        "         like TensorFlow\n",
        "-   `local_runner.py` — define a runner for local environment which uses\n",
        "local orchestration engine\n",
        "-   `kubeflow_runner.py` — define a runner for Kubeflow Pipelines\n",
        "orchestration engine\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "saF1CpVefaf1"
      },
      "source": [
        "By default, the template only includes standard TFX components. If you need\n",
        "some customized actions, you can create custom components for your pipeline.\n",
        "Please see\n",
        "[TFX custom component guide](https://www.tensorflow.org/tfx/guide/understanding_custom_components)\n",
        "for the detail."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ROwHAsDK0olT"
      },
      "source": [
        "#### Unit-test files.\n",
        "\n",
        "You might notice that there are some files with `_test.py` in their name.\n",
        "These are unit tests of the pipeline and it is recommended to add more unit\n",
        "tests as you implement your own pipelines.\n",
        "You can run unit tests by supplying the module name of test files with `-m`\n",
        "flag. You can usually get a module name by deleting `.py` extension and\n",
        "replacing `/` with `.`.  For example:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "M0cMdE2Z0olU"
      },
      "outputs": [],
      "source": [
        "import sys\n",
        "!{sys.executable} -m models.features_test"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tO9Jhplo0olX"
      },
      "source": [
        "### Create a TFX pipeline in local environment.\n",
        "\n",
        "TFX supports several orchestration engines to run pipelines. We will use\n",
        "local orchestration engine. Local orchestration engine runs without any further dependencies, and it is suitable for development and debugging because it runs\n",
        "on local environment rather than depends on remote computing clusters."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ZO6yR8lOo5GZ"
      },
      "source": [
        "We will use `local_runner.py` to run your pipeline using local\n",
        "orchestrator. You have to create a pipeline before running it. You can create\n",
        "a pipeline with `pipeline create` command.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "_9unbcHlo7Yi"
      },
      "outputs": [],
      "source": [
        "!tfx pipeline create --engine=local --pipeline_path=local_runner.py"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "NrRL6R06o99S"
      },
      "source": [
        "`pipeline create` command registers your pipeline defined in `local_runner.py`\n",
        "without actually running it.\n",
        "\n",
        "You will run the created pipeline with `run create` command in following steps.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7p73589GbTFi"
      },
      "source": [
        "## Step 2. Ingest YOUR data to the pipeline.\n",
        "\n",
        "The initial pipeline ingests the penguin dataset which is included in the\n",
        "template. You need to put your data into the pipeline, and most TFX\n",
        "pipelines start with ExampleGen component."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "AEVwa28qtjwi"
      },
      "source": [
        "### Choose an ExampleGen\n",
        "\n",
        "Your data can be stored anywhere your pipeline can access, on either a local or distributed filesystem, or a query-able system. TFX provides various\n",
        "[`ExampleGen` components](https://www.tensorflow.org/tfx/guide/examplegen)\n",
        "to bring your data into a TFX pipeline. You can choose one from following\n",
        "example generating components.\n",
        "\n",
        "- CsvExampleGen: Reads CSV files in a directory. Used in\n",
        "[penguin example](https://github.com/tensorflow/tfx/tree/master/tfx/examples/penguin)\n",
        "and\n",
        "[Chicago taxi example](https://github.com/tensorflow/tfx/tree/master/tfx/examples/chicago_taxi_pipeline).\n",
        "- ImportExampleGen: Takes TFRecord files with TF Example data format. Used in\n",
        "[MNIST examples](https://github.com/tensorflow/tfx/tree/master/tfx/examples/mnist).\n",
        "- FileBasedExampleGen for\n",
        "[Avro](https://github.com/tensorflow/tfx/blob/master/tfx/components/example_gen/custom_executors/avro_executor.py)\n",
        "or\n",
        "[Parquet](https://github.com/tensorflow/tfx/blob/master/tfx/components/example_gen/custom_executors/parquet_executor.py)\n",
        "format.\n",
        "- [BigQueryExampleGen](https://www.tensorflow.org/tfx/api_docs/python/tfx/extensions/google_cloud_big_query/example_gen/component/BigQueryExampleGen):\n",
        "Reads data in Google Cloud BigQuery directly. Used in\n",
        "[Chicago taxi examples](https://github.com/tensorflow/tfx/tree/master/tfx/examples/chicago_taxi_pipeline).\n",
        "\n",
        "You can also create your own ExampleGen, for example, tfx includes\n",
        "[a custom ExecampleGen which uses Presto](https://github.com/tensorflow/tfx/tree/master/tfx/examples/custom_components/presto_example_gen)\n",
        "as a data source. See\n",
        "[the guide](https://www.tensorflow.org/tfx/guide/examplegen#custom_examplegen)\n",
        "for more information on how to use and develop custom executors.\n",
        "\n",
        "Once you decide which ExampleGen to use, you will need to modify the pipeline\n",
        "definition to use your data.\n",
        "\n",
        "1. Modify the `DATA_PATH` in `local_runner.py` and set it to the location of\n",
        "your files.\n",
        "  - If you have files in local environment, specify the path. This is the\n",
        "  best option for developing or debugging a pipeline.\n",
        "  - If the files are stored in GCS, you can use a path starting with\n",
        "  `gs://{bucket_name}/...`. Please make sure that you can access GCS from\n",
        "  your terminal, for example, using\n",
        "  [`gsutil`](https://cloud.google.com/storage/docs/gsutil).\n",
        "  Please follow\n",
        "  [authorization guide in Google Cloud](https://cloud.google.com/sdk/docs/authorizing)\n",
        "  if needed.\n",
        "  - If you want to use a Query-based ExampleGen like BigQueryExampleGen, you\n",
        "  need a Query statement to select data from the data source. There are a few\n",
        "  more things you need to set to use Google Cloud BigQuery as a data source.\n",
        "    - In `pipeline/configs.py`:\n",
        "      - Change `GOOGLE_CLOUD_PROJECT` and `GCS_BUCKET_NAME` to your GCP\n",
        "      project and bucket name. The bucket should exist before we run\n",
        "      the pipeline.\n",
        "      - Uncomment `BIG_QUERY_WITH_DIRECT_RUNNER_BEAM_PIPELINE_ARGS` variable.\n",
        "      - Uncomment and set `BIG_QUERY_QUERY` variable to\n",
        "      **your query statement**.\n",
        "    - In `local_runner.py`:\n",
        "      - Comment out `data_path` argument and uncomment `query` argument\n",
        "      instead in `pipeline.create_pipeline()`.\n",
        "    - In `pipeline/pipeline.py`:\n",
        "      - Comment out `data_path` argument and uncomment `query` argument\n",
        "      in `create_pipeline()`.\n",
        "      - Use\n",
        "      [BigQueryExampleGen](https://www.tensorflow.org/tfx/api_docs/python/tfx/extensions/google_cloud_big_query/example_gen/component/BigQueryExampleGen)\n",
        "      instead of CsvExampleGen.\n",
        "\n",
        "1. Replace existing CsvExampleGen to your ExampleGen class in\n",
        "`pipeline/pipeline.py`. Each ExampleGen class has different signature.\n",
        "Please see [ExampleGen component guide](https://www.tensorflow.org/tfx/guide/examplegen) for more detail. Don't forget to import required modules with\n",
        "`import` statements in `pipeline/pipeline.py`."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "xiAG6acjbl5U"
      },
      "source": [
        "The initial pipeline is consist of four components, `ExampleGen`,\n",
        "`StatisticsGen`, `SchemaGen` and `ExampleValidator`. We don't need to change\n",
        "anything for `StatisticsGen`, `SchemaGen` and `ExampleValidator`. Let's run the\n",
        "pipeline for the first time."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "tKOI48WumF7h"
      },
      "outputs": [],
      "source": [
        "# Update and run the pipeline.\n",
        "!tfx pipeline update --engine=local --pipeline_path=local_runner.py \\\n",
        " && tfx run create --engine=local --pipeline_name={PIPELINE_NAME}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "eFbYFNVbbzna"
      },
      "source": [
        "You should see \"Component ExampleValidator is finished.\" if the pipeline ran successfully."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "uuD5FRPAcOn8"
      },
      "source": [
        "### Examine output of the pipeline."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tL1wWoDh5wkj"
      },
      "source": [
        "TFX pipeline produces two kinds of output, artifacts and a\n",
        "[metadata DB(MLMD)](https://www.tensorflow.org/tfx/guide/mlmd) which contains\n",
        "metadata of artifacts and pipeline executions. The location to the output is\n",
        "defined in `local_runner.py`. By default, artifacts are stored under\n",
        "`tfx_pipeline_output` directory and metadata is stored as an sqlite database\n",
        "under `tfx_metadata` directory.\n",
        "\n",
        "You can use MLMD APIs to examine these outputs. First, we will define some\n",
        "utility functions to search output artifacts that were just produced."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "K0i_jTvOI8mv"
      },
      "outputs": [],
      "source": [
        "import tensorflow as tf\n",
        "import tfx\n",
        "from ml_metadata.proto import metadata_store_pb2\n",
        "from tfx.types import artifact_utils\n",
        "\n",
        "# TODO(b/171447278): Move these functions into TFX library.\n",
        "\n",
        "def get_latest_executions(store, pipeline_name, component_id = None):\n",
        "  \"\"\"Fetch all pipeline runs.\"\"\"\n",
        "  if component_id is None:  # Find entire pipeline runs.\n",
        "    run_contexts = [\n",
        "        c for c in store.get_contexts_by_type('run')\n",
        "        if c.properties['pipeline_name'].string_value == pipeline_name\n",
        "    ]\n",
        "  else:  # Find specific component runs.\n",
        "    run_contexts = [\n",
        "        c for c in store.get_contexts_by_type('component_run')\n",
        "        if c.properties['pipeline_name'].string_value == pipeline_name and\n",
        "           c.properties['component_id'].string_value == component_id\n",
        "    ]\n",
        "  # Pick the latest run context.\n",
        "  latest_context = max(run_contexts,\n",
        "                       key=lambda c: c.last_update_time_since_epoch)\n",
        "  return store.get_executions_by_context(latest_context.id)\n",
        "\n",
        "def get_latest_artifacts(store, pipeline_name, component_id = None):\n",
        "  \"\"\"Fetch all artifacts from latest pipeline execution.\"\"\"\n",
        "  executions = get_latest_executions(store, pipeline_name, component_id)\n",
        "\n",
        "  # Fetch all artifacts produced from the given executions.\n",
        "  execution_ids = [e.id for e in executions]\n",
        "  events = store.get_events_by_execution_ids(execution_ids)\n",
        "  artifact_ids = [\n",
        "      event.artifact_id for event in events\n",
        "      if event.type == metadata_store_pb2.Event.OUTPUT\n",
        "  ]\n",
        "  return store.get_artifacts_by_id(artifact_ids)\n",
        "\n",
        "def find_latest_artifacts_by_type(store, artifacts, artifact_type):\n",
        "  \"\"\"Get the latest artifacts of a specified type.\"\"\"\n",
        "  # Get type information from MLMD\n",
        "  artifact_type = store.get_artifact_type(artifact_type)\n",
        "  # Filter artifacts with type.\n",
        "  filtered_artifacts = [aritfact for aritfact in artifacts\n",
        "                        if aritfact.type_id == artifact_type.id]\n",
        "  # Convert MLMD artifact data into TFX Artifact instances.\n",
        "  return [artifact_utils.deserialize_artifact(artifact_type, artifact)\n",
        "      for artifact in filtered_artifacts]\n",
        "\n",
        "\n",
        "from tfx.orchestration.experimental.interactive import visualizations\n",
        "\n",
        "def visualize_artifacts(artifacts):\n",
        "  \"\"\"Visualizes artifacts using standard visualization modules.\"\"\"\n",
        "  for artifact in artifacts:\n",
        "    visualization = visualizations.get_registry().get_visualization(\n",
        "        artifact.type_name)\n",
        "    if visualization:\n",
        "      visualization.display(artifact)\n",
        "\n",
        "from tfx.orchestration.experimental.interactive import standard_visualizations\n",
        "standard_visualizations.register_standard_visualizations()\n",
        "\n",
        "import pprint\n",
        "\n",
        "from tfx.orchestration import metadata\n",
        "from tfx.types import artifact_utils\n",
        "from tfx.types import standard_artifacts\n",
        "\n",
        "def preview_examples(artifacts):\n",
        "  \"\"\"Preview a few records from Examples artifacts.\"\"\"\n",
        "  pp = pprint.PrettyPrinter()\n",
        "  for artifact in artifacts:\n",
        "    print(\"==== Examples artifact:{}({})\".format(artifact.name, artifact.uri))\n",
        "    for split in artifact_utils.decode_split_names(artifact.split_names):\n",
        "      print(\"==== Reading from split:{}\".format(split))\n",
        "      split_uri = os.path.join(artifact.uri, split)\n",
        "\n",
        "      # Get the list of files in this directory (all compressed TFRecord files)\n",
        "      tfrecord_filenames = [os.path.join(split_uri, name)\n",
        "                            for name in os.listdir(split_uri)]\n",
        "      # Create a `TFRecordDataset` to read these files\n",
        "      dataset = tf.data.TFRecordDataset(tfrecord_filenames,\n",
        "                                        compression_type=\"GZIP\")\n",
        "      # Iterate over the first 2 records and decode them.\n",
        "      for tfrecord in dataset.take(2):\n",
        "        serialized_example = tfrecord.numpy()\n",
        "        example = tf.train.Example()\n",
        "        example.ParseFromString(serialized_example)\n",
        "        pp.pprint(example)\n",
        "\n",
        "import local_runner\n",
        "\n",
        "metadata_connection_config = metadata.sqlite_metadata_connection_config(\n",
        "              local_runner.METADATA_PATH)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "cmwor9nVcmxy"
      },
      "source": [
        "Now we can read metadata of output artifacts from MLMD."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "TtsrZEUB1-J4"
      },
      "outputs": [],
      "source": [
        "with metadata.Metadata(metadata_connection_config) as metadata_handler:\n",
        "    # Search all aritfacts from the previous pipeline run.\n",
        "    artifacts = get_latest_artifacts(metadata_handler.store, PIPELINE_NAME)\n",
        "    # Find artifacts of Examples type.\n",
        "    examples_artifacts = find_latest_artifacts_by_type(\n",
        "        metadata_handler.store, artifacts,\n",
        "        standard_artifacts.Examples.TYPE_NAME)\n",
        "    # Find artifacts generated from StatisticsGen.\n",
        "    stats_artifacts = find_latest_artifacts_by_type(\n",
        "        metadata_handler.store, artifacts,\n",
        "        standard_artifacts.ExampleStatistics.TYPE_NAME)\n",
        "    # Find artifacts generated from SchemaGen.\n",
        "    schema_artifacts = find_latest_artifacts_by_type(\n",
        "        metadata_handler.store, artifacts,\n",
        "        standard_artifacts.Schema.TYPE_NAME)\n",
        "    # Find artifacts generated from ExampleValidator.\n",
        "    anomalies_artifacts = find_latest_artifacts_by_type(\n",
        "        metadata_handler.store, artifacts,\n",
        "        standard_artifacts.ExampleAnomalies.TYPE_NAME)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "3U5MNAUIdBtN"
      },
      "source": [
        "Now we can examine outputs from each component.\n",
        "[Tensorflow Data Validation(TFDV)](https://www.tensorflow.org/tfx/data_validation/get_started)\n",
        "is used in `StatisticsGen`, `SchemaGen` and `ExampleValidator`, and TFDV can\n",
        "be used to visualize outputs from these components.\n",
        "\n",
        "In this tutorial, we will use visualzation helper methods in TFX which use TFDV\n",
        "internally to show the visualization. Please see\n",
        "[TFX components tutorial](https://www.tensorflow.org/tfx/tutorials/tfx/components_keras)\n",
        "to learn more about each component."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "AxS6FsgU2IoZ"
      },
      "source": [
        "#### Examine output form ExampleGen\n",
        "\n",
        "Let's examine output from ExampleGen. Take a look at the first two examples for\n",
        "each split:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "3NWzXEcE13tW"
      },
      "outputs": [],
      "source": [
        "preview_examples(examples_artifacts)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Q0uiuPhkGEBz"
      },
      "source": [
        "By default, TFX ExampleGen divides examples into two splits, *train* and\n",
        "*eval*, but you can\n",
        "[adjust your split configuration](https://www.tensorflow.org/tfx/guide/examplegen#span_version_and_split)."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "yVh13wJu-IRv"
      },
      "source": [
        "#### Examine output from StatisticsGen\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "9LipxUp7-IRw"
      },
      "outputs": [],
      "source": [
        "visualize_artifacts(stats_artifacts)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "8aebEY4c0Ju7"
      },
      "source": [
        "These statistics are supplied to SchemaGen to construct a schema of data\n",
        "automatically."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ExEKbdw8-IRx"
      },
      "source": [
        "#### Examine output from SchemaGen\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "M2IURBSp-IRy"
      },
      "outputs": [],
      "source": [
        "visualize_artifacts(schema_artifacts)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "oTvj8yeBHDdU"
      },
      "source": [
        "This schema is automatically inferred from the output of StatisticsGen.\n",
        "We will use this generated schema in this tutorial, but you also can\n",
        "[modify and customize the schema](https://www.tensorflow.org/tfx/guide/statsgen#creating_a_curated_schema)."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "rl1PEUgo-IRz"
      },
      "source": [
        "#### Examine output from ExampleValidator\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "F-4oAjGR-IR0"
      },
      "outputs": [],
      "source": [
        "visualize_artifacts(anomalies_artifacts)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "t026ZzbU0961"
      },
      "source": [
        "If any anomalies were found, you may review your data that all examples\n",
        "follow your assumptions. Outputs from other components like StatistcsGen might\n",
        "be useful. Found anomalies don't block the pipeline execution."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "lFMmqy1W-IR1"
      },
      "source": [
        "You can see the available features from the outputs of the `SchemaGen`. If\n",
        "your features can be used to construct ML model in `Trainer` directly, you can\n",
        "skip the next step and go to Step 4. Otherwise you can do some feature\n",
        "engineering work in the next step. `Transform` component is needed when\n",
        "full-pass operations like calculating averages are required, especially when\n",
        "you need to scale."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "bYH8Y2KB0olm"
      },
      "source": [
        "## Step 3. (Optional) Feature engineering with Transform component.\n",
        "\n",
        "In this step, you will define various feature engineering job which will be\n",
        "used by `Transform` component in the pipeline. See\n",
        "[Transform component guide](https://www.tensorflow.org/tfx/guide/transform)\n",
        "for more information.\n",
        "\n",
        "This is only necessary if you training code requires additional feature(s)\n",
        "which is not available in the output of ExampleGen. Otherwise, feel free to\n",
        "fast forward to next step of using Trainer."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "qm_JjQUydbbb"
      },
      "source": [
        "### Define features of the model\n",
        "\n",
        "`models/features.py` contains constants to define features for the model\n",
        "including feature names, size of vocabulariy and so on. By default `penguin`\n",
        "template has two costants, `FEATURE_KEYS` and `LABEL_KEY`, because our `penguin`\n",
        "model solves a classification problem using supervised learning and all\n",
        "features are continuous numeric features. See\n",
        "[feature definitions from the chicago taxi example](https://github.com/tensorflow/tfx/blob/master/tfx/experimental/templates/taxi/models/features.py)\n",
        "for another example.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ATUeHXvJdcBn"
      },
      "source": [
        "### Implement preprocessing for training / serving in preprocessing_fn().\n",
        "\n",
        "Actual feature engineering happens in `preprocessing_fn()` function in\n",
        "`models/preprocessing.py`.\n",
        "\n",
        "In `preprocessing_fn` you can define a series of functions that manipulate the\n",
        "input dict of tensors to produce the output dict of tensors. There are helper\n",
        "functions like `scale_to_0_1` and `compute_and_apply_vocabulary` in the\n",
        "TensorFlow Transform API or you can simply use regular TensorFlow functions.\n",
        "By default `penguin` template includes example usages of\n",
        "[tft.scale_to_z_score](https://www.tensorflow.org/tfx/transform/api_docs/python/tft/scale_to_z_score)\n",
        "function to normalize feature values.\n",
        "\n",
        "See [Tensflow Transform guide](https://www.tensorflow.org/tfx/transform/get_started)\n",
        "for more information about authoring `preprocessing_fn`.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "xUg_Lc43dbTp"
      },
      "source": [
        "### Add Transform component to the pipeline.\n",
        "\n",
        "If your preprocessing_fn is ready, add `Transform` component to the pipeline.\n",
        "\n",
        "1. In `pipeline/pipeline.py` file, uncomment `# components.append(transform)`\n",
        "to add the component to the pipeline.\n",
        "\n",
        "You can update the pipeline and run again."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "VE-Pqvto0olm"
      },
      "outputs": [],
      "source": [
        "!tfx pipeline update --engine=local --pipeline_path=local_runner.py \\\n",
        " && tfx run create --engine=local --pipeline_name={PIPELINE_NAME}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "8q1ZYEHX0olo"
      },
      "source": [
        "If the pipeline ran successfully, you should see \"Component Transform is\n",
        "finished.\" *somewhere* in the log. Because `Transform` component and\n",
        "`ExampleValidator` component are not dependent to each other, the order of\n",
        "executions is not fixed. That said, either of `Transform` and\n",
        "`ExampleValidator` can be the last component in the pipeline execution."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XrPEnZt0E_0m"
      },
      "source": [
        "### Examine output from Transform\n",
        "\n",
        "Transform component creates two kinds of outputs, a Tensorflow graph and\n",
        "transformed examples. The transformed examples are Examples artifact type which\n",
        "is also produced by ExampleGen, but this one contains transformed feature\n",
        "values instead.\n",
        "\n",
        "You can examine them as we did in the previous step."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "FvC5S66ZU5g6"
      },
      "outputs": [],
      "source": [
        "with metadata.Metadata(metadata_connection_config) as metadata_handler:\n",
        "    # Search all aritfacts from the previous run of Transform component.\n",
        "    artifacts = get_latest_artifacts(metadata_handler.store,\n",
        "                                     PIPELINE_NAME, \"Transform\")\n",
        "    # Find artifacts of Examples type.\n",
        "    transformed_examples_artifacts = find_latest_artifacts_by_type(\n",
        "        metadata_handler.store, artifacts,\n",
        "        standard_artifacts.Examples.TYPE_NAME)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "CAFiEPKuC6Ib"
      },
      "outputs": [],
      "source": [
        "preview_examples(transformed_examples_artifacts)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "dWMBXU510olp"
      },
      "source": [
        "## Step 4. Train your model with Trainer component.\n",
        "\n",
        "We will build a ML model using `Trainer` component. See\n",
        "[Trainer component guide](https://www.tensorflow.org/tfx/guide/trainer)\n",
        "for more information. You need to provide your model code to the Trainer\n",
        "component.\n",
        "\n",
        "### Define your model.\n",
        "\n",
        "In penguin template, `models.model.run_fn` is used as `run_fn` argument for\n",
        "`Trainer` component. It means that `run_fn()` function in `models/model.py`\n",
        "will be called when `Trainer` component runs. You can see the code to construct\n",
        "a simple DNN model using `keras` API in given code. See\n",
        "[TensorFlow 2.x in TFX](https://www.tensorflow.org/tfx/guide/keras)\n",
        "guide for more information about using keras API in TFX.\n",
        "\n",
        "In this `run_fn`, you should build a model and save it to a directory pointed\n",
        "by `fn_args.serving_model_dir` which is specified by the component. You can use\n",
        "other arguments in `fn_args` which is passed into the `run_fn`. See\n",
        "[related codes](https://github.com/tensorflow/tfx/blob/b01482442891a49a1487c67047e85ab971717b75/tfx/components/trainer/executor.py#L141)\n",
        "for the full list of arguments in `fn_args`.\n",
        "\n",
        "Define your features in `models/features.py` and use them as needed. If you\n",
        "have transformed your features in Step 3, you should use transformed features\n",
        "as inputs to your model."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "oFiLIaCm-IR4"
      },
      "source": [
        "### Add Trainer component to the pipeline.\n",
        "\n",
        "If your run_fn is ready, add `Trainer` component to the pipeline.\n",
        "\n",
        "1. In `pipeline/pipeline.py` file, uncomment `# components.append(trainer)`\n",
        "to add the component to the pipeline.\n",
        "\n",
        "Arguments for the trainer component might depends on whether you use Transform\n",
        "component or not.\n",
        "- If you do **NOT** use `Transform` component, you don't need to change the\n",
        "arguments.\n",
        "- If you use `Transform` component, you need to change arguments\n",
        "when creating a `Trainer` component instance.\n",
        "  - Change `examples` argument to\n",
        "  `examples=transform.outputs['transformed_examples'],`. We need to use\n",
        "  transformed examples for training.\n",
        "  - Add `transform_graph` argument like\n",
        "  `transform_graph=transform.outputs['transform_graph'],`. This graph\n",
        "  contains TensorFlow graph for the transform operations.\n",
        "  - After above changes, the code for Trainer component creation will\n",
        "  look like following.\n",
        "\n",
        "  ```python\n",
        "  # If you use a Transform component.\n",
        "  trainer = Trainer(\n",
        "      run_fn=run_fn,\n",
        "      examples=transform.outputs['transformed_examples'],\n",
        "      transform_graph=transform.outputs['transform_graph'],\n",
        "      schema=schema_gen.outputs['schema'],\n",
        "      ...\n",
        "  ```\n",
        "\n",
        "You can update the pipeline and run again."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "VQDNitkH0olq"
      },
      "outputs": [],
      "source": [
        "!tfx pipeline update --engine=local --pipeline_path=local_runner.py \\\n",
        " && tfx run create --engine=local --pipeline_name={PIPELINE_NAME}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ksWfVQUnMYCX"
      },
      "source": [
        "When this execution runs successfully, you have now created and run your first\n",
        "TFX pipeline for your model. Congratulations!\n",
        "\n",
        "Your new model will be located in some place under the output directory, but it\n",
        "would be better to have a model in fixed location or service outside of the TFX\n",
        "pipeline which holds many interim results. Even better with continuous\n",
        "evaluation of the built model which is critical in ML production systems. We\n",
        "will see how continuous evaluation and deployments work in TFX in the next step."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "4DRTFdTy0ol3"
      },
      "source": [
        "## Step 5.  (Optional) Evaluate the model with Evaluator and publish with pusher.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "5DID2nzH-IR7"
      },
      "source": [
        "[`Evaluator`](https://www.tensorflow.org/tfx/guide/evaluator) component\n",
        "continuously evaluate every built model from `Trainer`, and\n",
        "[`Pusher`](https://www.tensorflow.org/tfx/guide/pusher) copies the model to\n",
        "a predefined location in the file system or even to\n",
        "[Google Cloud AI Platform Models](https://console.cloud.google.com/ai-platform/models).\n",
        "\n",
        "### Adds Evaluator component to the pipeline.\n",
        "\n",
        "In `pipeline/pipeline.py` file:\n",
        "1. Uncomment `# components.append(model_resolver)` to add latest model resolver\n",
        "to the pipeline. Evaluator can be used to compare a model with old baseline\n",
        "model which passed Evaluator in last pipeline run. `LatestBlessedModelResolver`\n",
        "finds the latest model which passed Evaluator.\n",
        "1. Set proper `tfma.MetricsSpec` for your model. Evaluation might be different\n",
        "for every ML model. In the penguin template, `SparseCategoricalAccuracy` was used\n",
        "because we are solving a multi category classification problem. You also need\n",
        "to specify `tfma.SliceSpec` to analyze your model for specific slices. For more\n",
        "detail, see\n",
        "[Evaluator component guide](https://www.tensorflow.org/tfx/guide/evaluator).\n",
        "1. Uncomment `# components.append(evaluator)` to add the component to the\n",
        "pipeline.\n",
        "\n",
        "You can update the pipeline and run again."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "i5_ojoZZmaDQ"
      },
      "outputs": [],
      "source": [
        "# Update and run the pipeline.\n",
        "!tfx pipeline update --engine=local --pipeline_path=local_runner.py \\\n",
        " && tfx run create --engine=local --pipeline_name={PIPELINE_NAME}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "apZX74qJ-IR7"
      },
      "source": [
        "### Examine output of Evaluator\n",
        "This step requires TensorFlow Model Analysis(TFMA) Jupyter notebook extension.\n",
        "Note that the version of the TFMA notebook extension should be identical to the\n",
        "version of TFMA python package.\n",
        "\n",
        "Following command will install TFMA notebook extension from NPM registry. It\n",
        "might take several minutes to complete."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "VoL46D5Pw5FX"
      },
      "outputs": [],
      "source": [
        "# Install TFMA notebook extension.\n",
        "!jupyter labextension install tensorflow_model_analysis@{tfma.__version__}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "GKMo4j8ww5PB"
      },
      "source": [
        "If installation is completed, please **reload your browser** to make the\n",
        "extension take effect."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "2ztotdqS-IR8"
      },
      "outputs": [],
      "source": [
        "with metadata.Metadata(metadata_connection_config) as metadata_handler:\n",
        "  # Search all aritfacts from the previous pipeline run.\n",
        "  artifacts = get_latest_artifacts(metadata_handler.store, PIPELINE_NAME)\n",
        "  model_evaluation_artifacts = find_latest_artifacts_by_type(\n",
        "      metadata_handler.store, artifacts,\n",
        "      standard_artifacts.ModelEvaluation.TYPE_NAME)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "tVojwMCuDJuk"
      },
      "outputs": [],
      "source": [
        "tfma_result = tfma.load_eval_result(model_evaluation_artifacts[0].uri)\n",
        "tfma.view.render_slicing_metrics(tfma_result)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "18tqjyHN-IR9"
      },
      "source": [
        "### Adds Pusher component to the pipeline.\n",
        "\n",
        "If the model looks promising, we need to publish the model.\n",
        "[Pusher component](https://www.tensorflow.org/tfx/guide/pusher)\n",
        "can publish the model to a location in the filesystem or to GCP AI Platform\n",
        "Models using\n",
        "[a custom executor](https://github.com/tensorflow/tfx/blob/master/tfx/extensions/google_cloud_ai_platform/pusher/executor.py).\n",
        "\n",
        "`Evaluator` component continuously evaluate every built model from `Trainer`,\n",
        "and [`Pusher`](https://www.tensorflow.org/tfx/guide/pusher) copies the model to\n",
        "a predefined location in the file system or even to\n",
        "[Google Cloud AI Platform Models](https://console.cloud.google.com/ai-platform/models).\n",
        "\n",
        "1. In `local_runner.py`, set `SERVING_MODEL_DIR` to a directory to publish.\n",
        "1. In `pipeline/pipeline.py` file, uncomment `# components.append(pusher)`\n",
        "to add Pusher to the pipeline.\n",
        "\n",
        "You can update the pipeline and run again."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "QH81d9FsrSXS"
      },
      "outputs": [],
      "source": [
        "# Update and run the pipeline.\n",
        "!tfx pipeline update --engine=local --pipeline_path=local_runner.py \\\n",
        " && tfx run create --engine=local --pipeline_name={PIPELINE_NAME}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_K6Z18tC-IR-"
      },
      "source": [
        "You should be able to find your new model at `SERVING_MODEL_DIR`."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "20KRGsPX0ol3"
      },
      "source": [
        "## Step 6.  (Optional) Deploy your pipeline to Kubeflow Pipelines on GCP.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "0X6vfy7s-IR-"
      },
      "source": [
        "As mentioned earlier, `local_runner.py` is good for debugging or development\n",
        "purpose but not a best solution for production workloads. In this step, we will\n",
        "deploy the pipeline to Kubeflow Pipelines on Google Cloud.\n",
        "\n",
        "### Preparation\n",
        "We need `kfp` python package and `skaffold` program to deploy a pipeline to a\n",
        "Kubeflow Pipelines cluster."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Ge1bMUtU-IR_"
      },
      "outputs": [],
      "source": [
        "import sys\n",
        "!{sys.executable} -m pip install --user --upgrade -q kfp==1.0.0\n",
        "\n",
        "# Download skaffold and set it executable.\n",
        "!curl -Lo skaffold https://storage.googleapis.com/skaffold/releases/latest/skaffold-linux-amd64 && chmod +x skaffold"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ZsbnGq52-ISB"
      },
      "source": [
        "You need to move `skaffold` binary to the place where your shell can find it.\n",
        "Or you can specify the path to skaffold when you run `tfx` binary with\n",
        "`--skaffold-cmd` flag."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "4amQ0Elz-ISC"
      },
      "outputs": [],
      "source": [
        "# Move skaffold binary into your path\n",
        "!mv skaffold /home/jupyter/.local/bin/"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "1rmyns-o-ISD"
      },
      "source": [
        "You also need a Kubeflow Pipelines cluster to run the pipeline. Please\n",
        "follow Step 1 and 2 in\n",
        "[TFX on Cloud AI Platform Pipelines tutorial](https://www.tensorflow.org/tfx/tutorials/tfx/cloud-ai-platform-pipelines).\n",
        "\n",
        "When your cluster is ready, open the pipeline dashboard by clicking\n",
        "*Open Pipelines Dashboard* in the\n",
        "[`Pipelines` page of the Google cloud console](http://console.cloud.google.com/ai-platform/pipelines).\n",
        "The URL of this page is `ENDPOINT` to request a pipeline run. The endpoint\n",
        "value is everything in the URL after the https://, up to, and including,\n",
        "googleusercontent.com. Put your endpoint to following code block.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "hGyj-Qa3-ISD"
      },
      "outputs": [],
      "source": [
        "ENDPOINT='' # Enter your ENDPOINT here."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "igTo05YI-ISF"
      },
      "source": [
        "To run our code in a Kubeflow Pipelines cluster, we need to pack our code into\n",
        "a container image. The image will be built automatically while deploying our\n",
        "pipeline, and you only need to set a name and an container registry for your\n",
        "image. In our example, we will use\n",
        "[Google Container registry](https://cloud.google.com/container-registry),\n",
        "and name it `tfx-pipeline`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "5J3LrI0K-ISF"
      },
      "outputs": [],
      "source": [
        "# Read GCP project id from env.\n",
        "shell_output=!gcloud config list --format 'value(core.project)' 2>/dev/null\n",
        "GOOGLE_CLOUD_PROJECT=shell_output[0]\n",
        "\n",
        "# Docker image name for the pipeline image.\n",
        "CUSTOM_TFX_IMAGE='gcr.io/' + GOOGLE_CLOUD_PROJECT + '/tfx-pipeline'"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Gg11pLmU-ISH"
      },
      "source": [
        "### Set data location.\n",
        "\n",
        "Your data should be accessible from the Kubeflow Pipelines cluster. If you have\n",
        "used data in your local environment, you might need to upload it to remote\n",
        "storage like Google Cloud Storage. For example, we can upload penguin data to a\n",
        "default bucket which is created automatically when a Kubeflow Pipelines cluster\n",
        "is deployed like following."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "y8MmRIHi-ISH"
      },
      "outputs": [],
      "source": [
        "!gsutil cp data/data.csv gs://{GOOGLE_CLOUD_PROJECT}-kubeflowpipelines-default/tfx-template/data/penguin/"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ASc1tDMm-ISJ"
      },
      "source": [
        "Update the data location stored at `DATA_PATH` in `kubeflow_runner.py`.\n",
        "\n",
        "If you are using BigQueryExampleGen, there is no need to upload the data file,\n",
        "but please make sure that `kubeflow_runner.py` uses the same `query` and\n",
        "`beam_pipeline_args` argument for `pipeline.create_pipeline()` function."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "q42gn3XS-ISK"
      },
      "source": [
        "### Deploy the pipeline.\n",
        "\n",
        "If everything is ready, you can create a pipeline using `tfx pipeline create`\n",
        "command.\n",
        "> Note: When creating a pipeline for Kubeflow Pipelines, we need a container\n",
        "image which will be used to run our pipeline. And `skaffold` will build the\n",
        "image for us. Because `skaffold` pulls base images from the docker hub, it will\n",
        "take 5~10 minutes when we build the image for the first time, but it will take\n",
        "much less time from the second build.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ytZ0liBn-ISK"
      },
      "outputs": [],
      "source": [
        "!tfx pipeline create  \\\n",
        "--engine=kubeflow \\\n",
        "--pipeline-path=kubeflow_runner.py \\\n",
        "--endpoint={ENDPOINT} \\\n",
        "--build-target-image={CUSTOM_TFX_IMAGE}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "fFqUQxQG-ISM"
      },
      "source": [
        "Now start an execution run with the newly created pipeline using the\n",
        "`tfx run create` command."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "4ps-4RHz-ISM"
      },
      "outputs": [],
      "source": [
        "!tfx run create --engine=kubeflow --pipeline-name={PIPELINE_NAME} --endpoint={ENDPOINT}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Fx3LtAL0-ISN"
      },
      "source": [
        "Or, you can also run the pipeline in the Kubeflow Pipelines dashboard. The new\n",
        "run will be listed under `Experiments` in the Kubeflow Pipelines dashboard.\n",
        "Clicking into the experiment will allow you to monitor progress and visualize\n",
        "the artifacts created during the execution run."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "mP8W6zjD-ISO"
      },
      "source": [
        "If you are interested in running your pipeline on Kubeflow Pipelines,\n",
        "find more instructions in\n",
        "[TFX on Cloud AI Platform Pipelines tutorial](https://www.tensorflow.org/tfx/tutorials/tfx/cloud-ai-platform-pipelines)."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "PTsgD_Kz-ISO"
      },
      "source": [
        "### Cleaning up\n",
        "\n",
        "To clean up all Google Cloud resources used in this step, you can\n",
        "[delete the Google Cloud project](https://cloud.google.com/resource-manager/docs/creating-managing-projects#shutting_down_projects)\n",
        "you used for the tutorial.\n",
        "\n",
        "Alternatively, you can clean up individual resources by visiting each\n",
        "consoles:\n",
        "- [Google Cloud Storage](https://console.cloud.google.com/storage)\n",
        "- [Google Container Registry](https://console.cloud.google.com/gcr)\n",
        "- [Google Kubernetes Engine](https://console.cloud.google.com/kubernetes)"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "collapsed_sections": [],
      "name": "penguin_template.ipynb",
      "toc_visible": true
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}
